﻿#include "kafka.h"
#include <QDebug>

#include <iostream>
#include <string>
#include <cstdlib>
#include <cstdio>
#include <signal.h>



static bool m_run = 1;
static bool exit_eof = false;
static int eof_cnt = 0;
static int partition_cnt = 0;
static int verbosity = 1;
static long msg_cnt = 0;
static int64_t msg_bytes = 0;
static void sigterm(int sig)
{
	m_run = 0;
}


kafka* g_pDlg = nullptr;



class ExampleEventCb : public RdKafka::EventCb
{
public:
	void event_cb(RdKafka::Event& event)
	{


		switch (event.type())
		{
		case RdKafka::Event::EVENT_ERROR:
			if (event.fatal())
			{
				std::cerr << "FATAL ";
				m_run = 0;
			}
			std::cerr << "ERROR (" << RdKafka::err2str(event.err())
				<< "): " << event.str() << std::endl;
			break;

		case RdKafka::Event::EVENT_STATS:
			std::cerr << "\"STATS\": " << event.str() << std::endl;
			break;

		case RdKafka::Event::EVENT_LOG:
			fprintf(stderr, "LOG-%i-%s: %s\n", event.severity(), event.fac().c_str(),
				event.str().c_str());
			break;

		case RdKafka::Event::EVENT_THROTTLE:
			std::cerr << "THROTTLED: " << event.throttle_time() << "ms by "
				<< event.broker_name() << " id " << (int)event.broker_id()
				<< std::endl;
			break;

		default:
			std::cerr << "EVENT " << event.type() << " ("
				<< RdKafka::err2str(event.err()) << "): " << event.str()
				<< std::endl;
			break;
		}
	}
};

class ExampleRebalanceCb : public RdKafka::RebalanceCb
{
private:
	static void part_list_print(
		const std::vector<RdKafka::TopicPartition*>& partitions)
	{
		for (unsigned int i = 0; i < partitions.size(); i++)
			std::cerr << partitions[i]->topic() << "[" << partitions[i]->partition()
			<< "], ";
		std::cerr << "\n";
	}

public:
	void rebalance_cb(RdKafka::KafkaConsumer* consumer,
		RdKafka::ErrorCode err,
		std::vector<RdKafka::TopicPartition*>& partitions)
	{
		std::cerr << "RebalanceCb: " << RdKafka::err2str(err) << ": ";

		part_list_print(partitions);

		RdKafka::Error* error = NULL;
		RdKafka::ErrorCode ret_err = RdKafka::ERR_NO_ERROR;

		if (err == RdKafka::ERR__ASSIGN_PARTITIONS)
		{
			if (consumer->rebalance_protocol() == "COOPERATIVE")
				error = consumer->incremental_assign(partitions);
			else
				ret_err = consumer->assign(partitions);
			partition_cnt += (int)partitions.size();
		}
		else
		{
			if (consumer->rebalance_protocol() == "COOPERATIVE")
			{
				error = consumer->incremental_unassign(partitions);
				partition_cnt -= (int)partitions.size();
			}
			else {
				ret_err = consumer->unassign();
				partition_cnt = 0;
			}
		}
		eof_cnt = 0; /* FIXME: Won't work with COOPERATIVE */

		if (error)
		{
			std::cerr << "incremental assign failed: " << error->str() << "\n";
			delete error;
		}
		else if (ret_err)
			std::cerr << "assign failed: " << RdKafka::err2str(ret_err) << "\n";
	}
};

// 消费者回调类
class ExampleConsumeCb : public RdKafka::ConsumeCb
{
public:
	// 通过 ConsumeCb 继承
	void consume_cb(RdKafka::Message& message, void* opaque)
	{
		g_pDlg->msg_consume(&message, opaque);
	}



};




kafka::kafka(std::string ip, QString strID ,QObject* parent) :
	QThread(parent)
{
	m_kafkaIp = ip;
	m_strId = strID;

	g_pDlg = this;
}

kafka::~kafka()
{

}

void kafka::msg_consume(RdKafka::Message* message, void* opaque)
{
	switch (message->err())
	{
	case RdKafka::ERR__TIMED_OUT:
		break;

	case RdKafka::ERR_NO_ERROR:
		/* Real message */
		msg_cnt++;
		msg_bytes += message->len();
		if (verbosity >= 3)
			std::cerr << "Read msg at offset " << message->offset() << std::endl;
		RdKafka::MessageTimestamp ts;
		ts = message->timestamp();
		if (verbosity >= 2 &&
			ts.type != RdKafka::MessageTimestamp::MSG_TIMESTAMP_NOT_AVAILABLE)
		{
			std::string tsname = "?";
			if (ts.type == RdKafka::MessageTimestamp::MSG_TIMESTAMP_CREATE_TIME)
				tsname = "create time";
			else if (ts.type ==
				RdKafka::MessageTimestamp::MSG_TIMESTAMP_LOG_APPEND_TIME)
				tsname = "log append time";
			std::cout << "Timestamp: " << tsname << " " << ts.timestamp << std::endl;
		}
		if (verbosity >= 2 && message->key())
		{
			std::cout << "Key: " << *message->key() << std::endl;
		}
		if (verbosity >= 1)
		{
// 			printf("%.*s\n", static_cast<int>(message->len()),
// 				static_cast<const char*>(message->payload()));


			QString strName(QString::fromStdString(message->topic_name()));
			QString strData(static_cast<const char*>(message->payload()));
			//数据在这里可以进行处理
			/*
				处理代码
			*/

			emit g_pDlg->valueUpdate(strName, strData);
		}
		break;

	case RdKafka::ERR__PARTITION_EOF:
		/* Last message */
		if (exit_eof && ++eof_cnt == partition_cnt)
		{
			std::cerr << "%% EOF reached for all " << partition_cnt << " partition(s)"
				<< std::endl;
			m_run = 0;
		}
		break;

	case RdKafka::ERR__UNKNOWN_TOPIC:
	case RdKafka::ERR__UNKNOWN_PARTITION:
		std::cerr << "Consume failed: " << message->errstr() << std::endl;
		m_run = 0;
		break;

	default:
		/* Errors */
		std::cerr << "Consume failed: " << message->errstr() << std::endl;
		m_run = 0;
	}
}

void kafka::stop()
{
	m_run = false;

	this->wait();
}

//重写线程执行函数
void kafka::run()
{

	std::string brokers = m_kafkaIp;

	std::string errstr;
	std::string mode;
	std::string debug;
	// 订阅topic
	std::vector<std::string> topics;

	topics.push_back("test-topic");

  	topics.push_back("battery_shell_broken");			// 动力电池蒙皮破损检测
 	topics.push_back("battery_kettle_broken");			// 动力电池水壶破损检测
 
 	topics.push_back("battery_top_invade");				// 动力电池顶部异物检测
 	topics.push_back("battery_frame_deform");			// 动力电池框架变形检测
 	topics.push_back("battery_tilt");					// 动力电池倾斜识别

	topics.push_back("jiuxing_steel_rope_broken");		// 玖行钢丝绳断股检测
	topics.push_back("jiuxing_rope_coiler_failure");	// 玖行卷绳器失效检测

 	topics.push_back("rongqing_steel_rope_broken");		// 融青钢丝绳断股检测
 	topics.push_back("rongqing_oil_leakage");			// 融青漏油检测

	bool do_conf_dump = false;


	RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);

	// Set configuration properties   
	conf->set("metadata.broker.list", brokers, errstr);

	QString str = m_strId + QString("_assets_missing_group_01");
	std::string strId = str.toUtf8().data();
	conf->set("group.id", strId, errstr);
	conf->set("enable.partition.eof", "true", errstr);


	ExampleRebalanceCb ex_rebalance_cb;
	conf->set("rebalance_cb", &ex_rebalance_cb, errstr);

	if (exit_eof) {
		std::string strategy;
		if (conf->get("partition.assignment.strategy", strategy) ==
			RdKafka::Conf::CONF_OK &&
			strategy == "cooperative-sticky") {
			std::cerr << "Error: this example has not been modified to " << "support -e (exit on EOF) when the partition.assignment.strategy "
				<< "is set to " << strategy << ": remove -e from the command line\n";
			exit(1);
		}
	}

	if (!debug.empty())
	{
		if (conf->set("debug", debug, errstr) != RdKafka::Conf::CONF_OK)
		{
			std::cerr << errstr << std::endl;
			exit(1);
		}
	}

	ExampleEventCb ex_event_cb;
	conf->set("event_cb", &ex_event_cb, errstr);

	// 异步版本
	ExampleConsumeCb ex_consume_cb;
	//RdKafka::Conf::ConfResult aa = conf->set("consume_cb", &ex_consume_cb, errstr);
	if (conf->set("consume_cb", &ex_consume_cb, errstr) != RdKafka::Conf::CONF_OK) {
		std::cerr << "Failed to set consume_cb: " << errstr << std::endl;
	}

	if (do_conf_dump) 
	{
		std::list<std::string>* dump;
		dump = conf->dump();
		std::cout << "# Global config" << std::endl;

		for (std::list<std::string>::iterator it = dump->begin(); it != dump->end();) 
		{
			std::cout << *it << " = ";
			it++;
			std::cout << *it << std::endl;
			it++;
		}
		std::cout << std::endl;

		exit(0);
	}

	signal(SIGINT, sigterm);
	signal(SIGTERM, sigterm);

	RdKafka::KafkaConsumer* consumer = RdKafka::KafkaConsumer::create(conf, errstr);
	if (!consumer) {
		std::cerr << "Failed to create consumer: " << errstr << std::endl;
		exit(1);
	}

	std::cout << "% Created consumer " << consumer->name() << std::endl;

	RdKafka::ErrorCode err = consumer->subscribe(topics);
	if (err) {
		std::cerr << "Failed to subscribe to " << topics.size() << " topics: " << RdKafka::err2str(err) << std::endl;
		exit(1);
	}

	/*
	 * Consume messages
	 */
	while (m_run)
	{
		consumer->poll(100);
		// msg_consume(msg, NULL);
	   //delete msg;
		//msleep(100);
	}

	consumer->close();
	consumer->unsubscribe();

	delete consumer;
	delete conf;

	std::cerr << "% Consumed " << msg_cnt << " messages (" << msg_bytes << " bytes)" << std::endl;

	/*
	 * Wait for RdKafka to decommission.
	 * This is not strictly needed (with check outq_len() above), but
	 * allows RdKafka to clean up all its resources before the application
	 * exits so that memory profilers such as valgrind wont complain about
	 * memory leaks.
	 */
	RdKafka::wait_destroyed(5000);
}



